fix: handle redis timeouts and cap concurrent backup dispatch - #95
fix: handle redis timeouts and cap concurrent backup dispatch#95RywJakkraphat wants to merge 3 commits into
Conversation
|
Warning Review limit reached
Next review available in: 5 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (9)
📝 WalkthroughWalkthroughChangesBackup concurrency limiting
Scheduler error handling
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
Cargo.toml (1)
22-22: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winDeclare Tokio’s
timefeature directly.
src/tests/services/backup_dispatcher_tests.rsimportstokio::time, but this feature list omits"time". Tokio 1.49.0 gates that module behind the separate feature, so add it explicitly rather than relying on another dependency to enable it. (docs.rs)Proposed change
-tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros", "fs", "sync"] } +tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros", "fs", "sync", "time"] }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@Cargo.toml` at line 22, Update the Tokio dependency declaration to include the "time" feature explicitly, preserving all existing features so tokio::time imports compile reliably.src/tests/services/backup_dispatcher_tests.rs (1)
25-65: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftExercise
BackupService::dispatch, not a second semaphore.This test creates an independent
Semaphoreand never callsdispatchorexecute_backup; it will pass even if production code stops acquiringBACKUP_SEMAPHORE. Add an instrumented dispatcher test or extract the gating logic into a testable helper.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/tests/services/backup_dispatcher_tests.rs` around lines 25 - 65, The test semaphore_gated_execution_never_exceeds_configured_limit currently validates an independent semaphore rather than production behavior. Rewrite it to invoke BackupService::dispatch with an instrumented backup execution path and assert the observed concurrency stays within the configured limit; alternatively, extract dispatch’s semaphore-acquire/hold/release logic into a helper and test that helper while preserving the production wiring through BACKUP_SEMAPHORE.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/settings.rs`:
- Around line 53-65: Update the MAX_CONCURRENT_BACKUPS parsing in the settings
initializer to trim before parsing and treat malformed values as None instead of
panicking; preserve the explicit panic for zero, and cap valid values at
Semaphore::MAX_PERMITS before BACKUP_SEMAPHORE invokes Semaphore::new.
In `@src/tests/services/backup_dispatcher_tests.rs`:
- Around line 12-22: Replace the process-wide BACKUP_SEMAPHORE assertion in
backup_semaphore_defaults_to_unlimited_when_max_concurrent_backups_unset with a
hermetic check: either test the underlying semaphore-construction function using
explicit unset input, or run the check in a subprocess after removing
MAX_CONCURRENT_BACKUPS. Do not depend on CONFIG or BACKUP_SEMAPHORE Lazy
initialization or inherited environment state.
In `@src/utils/task_manager/scheduler.rs`:
- Around line 68-75: Update the scheduling flow around the task dispatch and
rescheduling zadd so a failed reschedule cannot leave the already-executed task
due for another dispatch. Advance or claim the schedule durably before invoking
the backup side effect, while preserving task execution only for successfully
claimed schedule entries.
- Around line 105-108: Update execute_task to initialize the shared context
through a fallible Context::try_new() instead of Context::new(), propagating its
error before constructing ConfigService and BackupService. Add
Context::try_new() so missing or invalid EDGE_KEY returns an error rather than
panicking, while preserving the existing context setup for valid keys.
---
Nitpick comments:
In `@Cargo.toml`:
- Line 22: Update the Tokio dependency declaration to include the "time" feature
explicitly, preserving all existing features so tokio::time imports compile
reliably.
In `@src/tests/services/backup_dispatcher_tests.rs`:
- Around line 25-65: The test
semaphore_gated_execution_never_exceeds_configured_limit currently validates an
independent semaphore rather than production behavior. Rewrite it to invoke
BackupService::dispatch with an instrumented backup execution path and assert
the observed concurrency stays within the configured limit; alternatively,
extract dispatch’s semaphore-acquire/hold/release logic into a helper and test
that helper while preserving the production wiring through BACKUP_SEMAPHORE.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 0808691f-3f39-4382-9713-056fa5bfdcfe
📒 Files selected for processing (9)
Cargo.tomlsrc/services/backup/dispatcher.rssrc/settings.rssrc/tests/services/backup_dispatcher_tests.rssrc/tests/services/mod.rssrc/tests/utils/mod.rssrc/tests/utils/task_manager_tests.rssrc/utils/task_manager/cron.rssrc/utils/task_manager/scheduler.rs
| let max_concurrent_backups = match env::var("MAX_CONCURRENT_BACKUPS") { | ||
| Ok(val) if val.trim().is_empty() => None, | ||
| Ok(val) => { | ||
| let parsed = val | ||
| .parse::<usize>() | ||
| .expect("MAX_CONCURRENT_BACKUPS must be a valid positive integer"); | ||
| if parsed == 0 { | ||
| panic!("MAX_CONCURRENT_BACKUPS must be at least 1"); | ||
| } | ||
| Some(parsed) | ||
| } | ||
| Err(_) => None, | ||
| }; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate settings.rs =="
fd -a 'settings\.rs$' . | sed 's#^\./##'
echo "== relevant settings.rs lines =="
if [ -f src/settings.rs ]; then
cat -n src/settings.rs | sed -n '1,140p'
fi
echo "== imports/usages for MAX_CONCURRENT_BACKUPS and CONFIG =="
rg -n "MAX_CONCURRENT_BACKUPS|CONFIG|Semaphore|MAX_PERMITS" .
echo "== Cargo/Tokio version references =="
for f in Cargo.toml Cargo.lock; do
if [ -f "$f" ]; then
echo "--- $f ---"
rg -n "tokio|MAX_PERMITS" "$f" || true
fi
done
echo "== Rust parse + bound behavioral probe =="
python3 - <<'PY'
values = ["", " ", "2", " 2 ", "0", "abc", str(2**64), str(2**32), str(42)]
for v in values:
ok_trim = None
try:
parsed = int(v)
ok_trim = parsed
except ValueError:
ok_trim = "parse-error"
ok_trim_trim = None
try:
parsed = int(v.strip())
ok_trim_trim = parsed
except ValueError:
ok_trim_trim = "parse-error"
print(f"{v!r}: parse={ok_trim}, parse(trim)={ok_trim_trim}")
PYRepository: Portabase/agent
Length of output: 9929
🌐 Web query:
Tokio Semaphore::new panic MAX_PERMITS documentation
💡 Result:
In the Tokio library, the Semaphore::new function will panic if the number of requested permits exceeds the constant Semaphore::MAX_PERMITS [1][2]. The constant Semaphore::MAX_PERMITS is defined as usize::MAX >> 3 [1][3]. This limit is enforced to ensure the internal state of the semaphore remains valid; attempts to exceed this limit, whether during initialization via Semaphore::new or when adding permits via Semaphore::add_permits, will trigger a panic [1][4].
Citations:
- 1: https://docs.rs/tokio/latest/tokio/sync/struct.Semaphore.html
- 2: https://docs.rs/tokio/1/tokio/sync/struct.Semaphore.html
- 3: https://docs.rs/tokio/1.50.0/tokio/sync/struct.Semaphore.html
- 4: https://docs.rs/tokio/latest/src/tokio/sync/semaphore.rs.html
Make MAX_CONCURRENT_BACKUPS parsing non-panicking.
parse().expect(...) panics on any unparseable input—including " 2 "—before the CONFIG static initializer even finishes. Use val.trim().parse::<usize>(), ignore malformed values by returning None, keep the explicit zero panic, and cap accepted values against Semaphore::MAX_PERMITS before BACKUP_SEMAPHORE calls Semaphore::new.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/settings.rs` around lines 53 - 65, Update the MAX_CONCURRENT_BACKUPS
parsing in the settings initializer to trim before parsing and treat malformed
values as None instead of panicking; preserve the explicit panic for zero, and
cap valid values at Semaphore::MAX_PERMITS before BACKUP_SEMAPHORE invokes
Semaphore::new.
| #[tokio::test] | ||
| async fn backup_semaphore_defaults_to_unlimited_when_max_concurrent_backups_unset() { | ||
| init_tracing_for_test(); | ||
|
|
||
| // Neither this test suite nor docker-compose.test.yml sets MAX_CONCURRENT_BACKUPS, | ||
| // so the real, process-wide BACKUP_SEMAPHORE must be None: dispatch() must not | ||
| // throttle backups unless an operator explicitly opts in. | ||
| assert!( | ||
| BACKUP_SEMAPHORE.is_none(), | ||
| "expected no concurrency cap by default (MAX_CONCURRENT_BACKUPS unset)" | ||
| ); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Make the default-behavior test hermetic.
CONFIG and BACKUP_SEMAPHORE are process-wide Lazy values, so this assertion can fail when the test runner inherits MAX_CONCURRENT_BACKUPS or another test initializes configuration first. Test a pure constructor with explicit inputs, or isolate this check in a subprocess with the variable removed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/tests/services/backup_dispatcher_tests.rs` around lines 12 - 22, Replace
the process-wide BACKUP_SEMAPHORE assertion in
backup_semaphore_defaults_to_unlimited_when_max_concurrent_backups_unset with a
hermetic check: either test the underlying semaphore-construction function using
explicit unset input, or run the check in a subprocess after removing
MAX_CONCURRENT_BACKUPS. Do not depend on CONFIG or BACKUP_SEMAPHORE Lazy
initialization or inherited environment state.
| let result: redis::RedisResult<()> = | ||
| conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await; | ||
| if let Err(e) = result { | ||
| error!( | ||
| "Failed to reschedule task={} key={}: {:?}", | ||
| task_clone.task, key, e | ||
| ); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Prevent duplicate backups after rescheduling fails.
The task has already executed when zadd fails. Its old score remains due, so the next scheduler tick dispatches the same backup again. Claim/advance the schedule successfully before starting the side effect, or otherwise add durable idempotency.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/utils/task_manager/scheduler.rs` around lines 68 - 75, Update the
scheduling flow around the task dispatch and rescheduling zadd so a failed
reschedule cannot leave the already-executed task due for another dispatch.
Advance or claim the schedule durably before invoking the backup side effect,
while preserving task execution only for successfully claimed schedule entries.
| let ctx = Arc::new(Context::new()); | ||
| let config_service = ConfigService::new(ctx.clone()); | ||
| let backup_service = BackupService::new(ctx.clone()); | ||
| let config = config_service.load(None).unwrap(); | ||
| let config = config_service.load(None).map_err(|e| anyhow::anyhow!(e))?; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Make context initialization fallible too.
Context::new() still panics when EDGE_KEY is absent or invalid, before config_service.load(None) can return an error. Add a fallible Context::try_new() and propagate it from execute_task.
Proposed direction
- let ctx = Arc::new(Context::new());
+ let ctx = Arc::new(Context::try_new()?);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/utils/task_manager/scheduler.rs` around lines 105 - 108, Update
execute_task to initialize the shared context through a fallible
Context::try_new() instead of Context::new(), propagating its error before
constructing ConfigService and BackupService. Add Context::try_new() so missing
or invalid EDGE_KEY returns an error rather than panicking, while preserving the
existing context setup for valid keys.
- settings.rs: extract parse_max_concurrent_backups as a pure, pub(crate) function. Trims whitespace before parsing (was panicking on values like " 2 "), and panics at boot if the value exceeds Semaphore::MAX_PERMITS instead of panicking later inside BACKUP_SEMAPHORE's Lazy init on first dispatch. Malformed/zero values still panic rather than silently falling back to unlimited - deliberate, matches this file's existing fail-fast convention for POOLING/CHUNK_SIZE_MB. - settings.rs: add 7 hermetic unit tests for the extracted function, colocated in a #[cfg(test)] mod since it's a pure function, not process/Redis-dependent like the rest of src/tests/. - dispatcher.rs: extract run_with_permit() so the semaphore acquire/hold/release logic is shared between dispatch() and its test, instead of the test validating a parallel reimplementation. Verified this actually catches a regression: temporarily made run_with_permit a no-op and confirmed the test fails. - Cargo.toml: declare tokio's "time" feature explicitly (used by scheduler.rs and the new tests) rather than relying on transitive enablement from another dependency, matching the same reasoning already applied to "sync". Two other CodeRabbit findings (potential duplicate dispatch if zadd reschedule fails after a task executes; Context::new() still panicking on bad EDGE_KEY inside execute_task) are real but out of scope for this fix: both require larger design changes (schedule-claim ordering, a new fallible Context::try_new() touching multiple call sites) rather than a surgical correction, and neither is a regression introduced by this PR. Left as follow-up work.
24a7e29 to
8b86b98
Compare
1 participant
Summary
Scheduler and cron sync used
.unwrap()on Redis reads/writes and JSONdeserialization, so a Redis timeout under CPU load panicked the process
and crash-looped. Periodic backup dispatch also had no concurrency limit,
letting jobs sharing a cron timestamp saturate host CPU and trigger those
same Redis timeouts.
scheduler.rs/cron.rs: replace the panicking.unwrap()calls withlogged error handling, matching this codebase's existing
match-and-log convention.
dispatcher.rs: gateexecute_backupbehind an optionaltokio::sync::Semaphore, configurable via the newMAX_CONCURRENT_BACKUPSenv var (not yet documented elsewhere in the repo — flagging here).
Unset = unlimited (unchanged from current behavior), so this is
opt-in and doesn't change default throughput on upgrade.
Since
dispatch()returns as soon as it spawns the backup task, asaturated semaphore can never delay the cron reschedule in
scheduler_loop— a full queue can't cause missed cron slots.Refs #94
Tests
execute_task's missing-args error pathcheck_and_update_cronand
scheduler_loophandling malformed Redis data without panickingBACKUP_SEMAPHOREis unlimited by default and thatits acquire/hold/release pattern actually caps concurrency
Every test was verified against the pre-fix code (temporarily reverted)
to confirm it fails at the original bug, then re-verified passing.
Test plan
cargo check/cargo clippy/cargo fmt --check— all cleancargo test— all new and existing tests passSummary by CodeRabbit
New Features
Bug Fixes
Tests